package com.etlmaster.executor.processors.Table;

import java.util.Date;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.DpExecutorUtils;
import com.etlmaster.executor.utils.HiveUtil;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * 创建表执行器
 * 
 * @author Wugf
 * @date 2016-03-08
 */
public abstract class CreateTableProcessor {
	protected Map<String, Object> cfgObj;
	protected JdbcTemplate jt;
	protected String ddl;
	protected CreateTableMeta meta;
	protected boolean isExist;
	protected JdbcTemplate jdbcTemplate;
	protected String dropDate;
	protected String schema;

	public abstract void buildDDl();

	public void initMetaInfo(CreateTableMeta meta, Map<String, Object> variables)
			throws Exception {
		jt = new JdbcTemplate();
		this.meta = meta;
		setMetaInfo();

		
		if (variables != null && variables.size() > 0)
			this.meta.variables.putAll(variables);
	}

	protected void setMetaInfo() throws Exception {
		cfgObj = DataCollectionUtils.getCreateTableMeta(
				meta.getXmlId(), jt);
		if(meta.getMetaTableName()!=null&&meta.getMetaTableName().length()>0)
			meta.setMetaTableName((String) cfgObj.get("dataname"));
		if (cfgObj == null)
			throw new Exception();
		if (cfgObj.get("creater") != null && cfgObj.get("eff_date") != null) {
			meta.setCreater(cfgObj.get("creater").toString());
			meta.setEff_date(cfgObj.get("eff_date").toString());
		} else {
			LogWriter.addLog("WARN", "元模型没有创建者或生效日期。");
		}
		if(cfgObj.get("dbname")!=null)
			meta.setDsName(cfgObj.get("dbname").toString());
	}

	public String execCrtTab() {
		checkTableExist();
		try {
			if (isExist) {
				if ("true".equals(meta.getOverride())) {
					if (!meta.getDsType().equals("hive")) {
						LogWriter.addLog("INFO", "表已经存在,删除原表{}",
								meta.getTableName());
						jdbcTemplate
								.update("drop table " + meta.getTableName());
						jdbcTemplate.batchUpdate(ddl.split(";"));
					} else {
						jdbcTemplate.execute("drop table "
								+ meta.getTableName());
						jdbcTemplate.execute(ddl);
					}
					LogWriter.addLog("INFO", "建表成功DDL: {}", ddl);
				} else {
					LogWriter.addLog("INFO", "表已经存在忽略建表");
				}
			} else {
				if (!meta.getDsType().equals("hive"))
					jdbcTemplate.batchUpdate(ddl.split(";"));
				else
					jdbcTemplate.execute(ddl);
				LogWriter.addLog("INFO", "建表成功DDL: {}", ddl);
			}
			return StepState.RUNNING_SUCCESSED + "";
		} catch (Exception e) {
			LogWriter.addLog("ERROR", "建表失败DDL:{} ", ddl);
			LogWriter.addLog("ERROR", "错误信息：{}", e.getMessage());
			return "建表失败DDL:{} " + ddl + " " + e.getMessage();
		}
	}

	public void checkTableExist() {
		jdbcTemplate = new JdbcTemplate(meta.getDsName());

		LogWriter.addLog("INFO", "dsType:{}", meta.getDsType());
		if (!meta.getDsType().equals("hive")) {
			try {
				jdbcTemplate.queryForObject(
						"select count(1) from " + meta.getTableName()
								+ " where 1=2", Long.class);
				isExist = true;
			} catch (Exception e) {
				LogWriter.addLog("WARN", "表[{}]不存在", meta.getTableName());
				LogWriter.addLog("ERROR", "错误信息：{}", e.getMessage());
			}
		} else {
			String sql = "desc " + meta.getTableName();
			isExist = HiveUtil.query(jdbcTemplate, sql);
		}

	}
/*
	private void setDbUserInfo(){
		Map<String,Object> map=DataCollectionUtils.fetchDBUserInfoByTeam(meta.getDsName(), meta.getTeamCode(), jt);
		meta.setUserName(map.get("username").toString());
		meta.setPassword(map.get("password").toString());
	}
	*/
	public String process(CreateTableMeta meta, Map<String, Object> variables) {
		try {
			getDdl(meta, variables);
		} catch (Exception e) {
			return StepState.RUNNING_FAILED + " " + e.getMessage();
		}

		DataSourceDefine def = DataSourceHolder.getInstance()
				.getDataSourceDefine(meta.getDsName());
		if(def==null){
			LogWriter.addLog("ERROR","未分配数据库资源！");
			return "未分配数据库资源！";
		}
		String resultState = execCrtTab();
		if (resultState.equals(StepState.RUNNING_SUCCESSED + "")) {
			if (cfgObj.get("DWDBDURA") != null && cfgObj.get("DATEFMT") != null) {
				dropDate = getDropDate(
						Integer.parseInt(cfgObj.get("DWDBDURA").toString()),
						cfgObj.get("DATEFMT").toString());
			} else {
				LogWriter.addLog("DEBUG", "表{}没设置存储周期", meta.getXmlId());

			}
			writeCrtTableLog(dropDate, jt);
		}

		return resultState;
	}

	public String getDdl(){
		return ddl;
	}
	public String getDdl(CreateTableMeta meta, Map<String, Object> variables)
			throws Exception {

		initMetaInfo(meta, variables);
		buildDDl();
		return ddl;
	}

	private void writeCrtTableLog(String dropDate, JdbcTemplate jdbcTemplate) {
		try {
			String sql = "delete from "+DataCollectionUtils.getProperties("tableall")+" where DBNAME=? and dataname=?  and xmlid=?";
			jdbcTemplate.update(
					sql,
					new Object[] { meta.getDsName(), meta.getTableName(),
							meta.getXmlId() });
			if (dropDate == null)
				dropDate = "99991231";
			sql = "INSERT INTO "+DataCollectionUtils.getProperties("tableall")+" (DBNAME,DATANAME,SCHEMA_NAME,TABSPACE,XMLID,MODELTAB,CREATOR,TASKID,EFF_DATE,dropdate)"
					+ "VALUES(?,?,?,?,?,?,?,?,?,?)";
			jdbcTemplate.update(
					sql,
					new Object[] { meta.getDsName(), meta.getTableName(),
							this.schema, meta.getTbsSpace(), meta.getXmlId(),
							meta.getMetaTableName(), meta.getCreater(),
							meta.getTaskid(), getCurrentDate("yyyyMMdd"),
							dropDate });

		} catch (Exception e) {
			LogWriter.addLog("ERROR", "{}", e.getMessage());
			LogWriter.addLog("ERROR", "记录建表日志失败{}", meta.getDsName(),
					meta.getTableName(), null, null, meta.getXmlId(),
					meta.getTableName(), meta.getUserName(), null);
		}
	}

	public static String getDropDate(int dura, String datefmt) {
		if (datefmt.equalsIgnoreCase("n") || !(dura > 0)) {
			return null;
		}
		String currentDate = getCurrentDate("yyyyMMdd");
		Map<String, Object> map = new HashMap<String, Object>();
		map.put("currentDate", currentDate);
		String freemarker = "${currentDate?calDate('" + dura + "','" + datefmt
				+ "')}";
		return DpExecutorUtils.variableSubstitution(freemarker, map);
	}

	public static String getCurrentDate(String datefmt) {
		SimpleDateFormat sdf = new SimpleDateFormat(datefmt);
		String date = sdf.format(new Date());
		return date;
	}

}
